package com.xuecheng.media.jobhandler;

import com.alibaba.nacos.client.naming.utils.CollectionUtils;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.j256.simplemagic.logger.Logger;
import com.j256.simplemagic.logger.LoggerFactory;
import com.xuecheng.base.exception.XueChengPlusException;
import com.xuecheng.base.utils.FFMpegUtils;
import com.xuecheng.content.checkutils.CheckInsertResultUtil;
import com.xuecheng.content.model.dto.MediaFilesDto;
import com.xuecheng.content.model.dto.MediaFilesTaskDto;
import com.xuecheng.content.model.dto.MultPartFileInfoInit;
import com.xuecheng.content.model.message.VideoProcessMessage;
import com.xuecheng.content.model.po.MediaFiles;
import com.xuecheng.content.model.po.MediaProcess;
import com.xuecheng.content.model.po.MediaProcessHistory;
import com.xuecheng.media.client.TeachplanMediaClient;
import com.xuecheng.media.config.threadpool.ThreadPoolAutoConfiguration;
import com.xuecheng.media.conver.MediaFilesConver;
import com.xuecheng.media.conver.VideoProcessMessageConver;
import com.xuecheng.media.conver.VideoTaskCover;
import com.xuecheng.media.mapper.MediaProcessHistoryMapper;
import com.xuecheng.media.mapper.MediaProcessMapper;
import com.xuecheng.media.service.MediaFilesService;
import com.xuecheng.media.service.MediaProcessHistoryService;
import com.xuecheng.media.service.MediaProcessService;
import com.xuecheng.media.service.MinIoService;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.DigestUtils;

import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;

import static org.springframework.boot.Banner.Mode.LOG;

/**
 * 视频处理任务整合类（线程池 + RocketMQ）
 */
@Slf4j
@Component
public class ThreadPoolMessageVideoTask {

    // ===================== 服务依赖 =====================
    @Autowired
    private MediaFilesService mediaFilesService;

    @Autowired
    private MediaProcessService mediaProcessService;

    @Autowired
    private MediaProcessHistoryService mediaProcessHistoryService;

    @Autowired
    private TeachplanMediaClient teachplanMediaClient;

    @Autowired
    private MinIoService minIoService;

    // ===================== RocketMQ 依赖 =====================
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    // ===================== 线程池依赖 =====================
    @Autowired
    private ThreadPoolExecutor threadPoolExecutor;

    // 缓存当前处理中的任务（避免重复处理）
    private final Set<String> processingFiles = ConcurrentHashMap.newKeySet();
    @Autowired
    private ObjectMapper objectMapper; // JSON处理器
    private static final Logger LOG = LoggerFactory.getLogger(ThreadPoolMessageVideoTask.class);

    // ===================== XXL-JOB 任务入口 =====================

    /**
     * XXL-JOB 视频处理调度任务
     */
    @XxlJob("convertedMp4Video")
    public void discoverAndDispatch() {
        XxlJobHelper.log("开始扫描待处理视频");

        try {
            // 1. 查询待处理视频（从数据库获取）
            List<MediaFilesTaskDto> pendingVideos = mediaFilesService.selectWaitConvertedVideos();

            if (CollectionUtils.isEmpty(pendingVideos)) {
                XxlJobHelper.log("当前无待处理视频");
                return;
            }

            XxlJobHelper.log("发现待处理视频数量: " + pendingVideos.size());

            // 2. 过滤掉正在处理的视频
            List<MediaFilesTaskDto> toProcess = pendingVideos.stream()
                    .filter(video -> !processingFiles.contains(video.getFileId()))
                    .collect(Collectors.toList());

            if (toProcess.isEmpty()) {
                XxlJobHelper.log("所有待处理视频已在处理中");
                return;
            }

            // 3. 发送到消息队列
            toProcess.forEach(this::dispatchForProcessing);

            XxlJobHelper.log("已发送视频数量: " + toProcess.size());

        } catch (Exception e) {
            String errorMsg = "视频调度失败: " + e.getMessage();
            XxlJobHelper.log(errorMsg);
            log.error(errorMsg, e);
            XxlJobHelper.handleFail(errorMsg);
        }
    }

    /**
     * 分发视频到处理队列
     */
    private void dispatchForProcessing(MediaFilesTaskDto taskDto) {
        try {
            processingFiles.add(taskDto.getFileId());

            // 创建消息对象
            VideoProcessMessage msg = new VideoProcessMessage(
                    taskDto.getFileId(),
                    taskDto.getUrl(),
                    taskDto.getFilename(),
                    taskDto.getBucket()
            );

            // 使用统一序列化方式
            String jsonMessage = objectMapper.writeValueAsString(msg);

            // 构建消息
            Message<String> message = MessageBuilder.withPayload(jsonMessage)
                    .setHeader(RocketMQHeaders.KEYS, taskDto.getFileId())
                    .setHeader(RocketMQHeaders.TAGS, "VIDEO_PROCESS")
                    .build();

            // 发送消息
            SendResult result = rocketMQTemplate.syncSend("VIDEO_PROCESS_TOPIC", message);

            LOG.info("消息发送成功 | 文件ID: {} | 消息ID: {}", taskDto.getFileId(), result.getMsgId());
        } catch (Exception e) {
            processingFiles.remove(taskDto.getFileId());
            LOG.error("消息发送失败 | 文件ID: {}", taskDto.getFileId(), e);
        }
    }

    // ===================== RocketMQ 消息处理 =====================

    /**
     * 处理视频处理请求（由监听器调用）
     */
    public void handleProcessingRequest(VideoProcessMessage message) {
        log.info("处理视频请求 | 文件ID: {}", message.getFileId());

        // 提交到线程池
        threadPoolExecutor.execute(() -> {
            try {
                processVideoFile(message);
            } catch (Exception e) {
                log.error("视频处理异常 | 文件ID: {}", message.getFileId(), e);
                handleProcessingError(message, e);
            } finally {
                // 从处理中集合移除
                processingFiles.remove(message.getFileId());
            }
        });
    }

    // ===================== 核心视频处理逻辑 =====================

    /**
     * 核心处理流程
     */
    @Transactional(rollbackFor = Exception.class)
    public void processVideoFile(VideoProcessMessage message) {
        final String fileId = message.getFileId();
        log.info("开始处理视频 | 文件ID: {}", fileId);

        // 1. 创建处理记录（需在实际处理逻辑中实现）
        MediaProcess task = createMediaProcess(message);

        try {
            // 2. 下载原始文件
            String originalPath = minIoService.downloadToTemp(message.getOriginalUrl());
            File originalFile = new File(originalPath);

            // 3. 转换文件格式
            String convertedPath = FFMpegUtils.convertToMp4(originalPath);
            File convertedFile = new File(convertedPath);

            // 4. 上传转换后文件 分块上传 并能初始化新文件信息
            MultPartFileInfoInit fileInfo = minIoService.splitFileToChunks(convertedFile);
            fileInfo.setFileName(task.getFilename());
            minIoService.mergeChunks(fileInfo);

            // 5. 创建新文件记录（需在实际服务中实现）
            MediaFiles newFile = mediaFilesService.convertedToMediaFile(fileInfo);
            CheckInsertResultUtil.checkExist(mediaFilesService.insert(newFile),()-> new XueChengPlusException("创建文件信息失败"));

            // 6. 更新关联数据
            updateReferences(task, newFile);

            // 7. 清理资源
            deleteTempFiles(originalFile, convertedFile); //清除临时文件
            mediaFilesService.deleteMinioById(task.getFileId());//清除minio记录 以及 数据库记录
            mediaProcessService.deleteByFileId(task.getFileId()); //删除正在处理的任务记录

            log.info("视频处理成功 | 文件ID: {}", fileId);

        } catch (Exception e) {
            log.error("视频处理失败 | 文件ID: {}", fileId, e);
            throw new RuntimeException("视频处理失败", e);
        }
    }

    // ===================== 辅助方法 =====================

    /**
     * 创建处理任务记录
     */
    private MediaProcess createMediaProcess(VideoProcessMessage message) {
        MediaProcess task = new MediaProcess();
        task.setFileId(message.getFileId());
        task.setUrl(message.getOriginalUrl());
        task.setFilename(message.getFilename());
        task.setBucket(message.getBucket());
        task.setStatus("1"); // 处理中
        task.setCreateDate(LocalDateTime.now());
        mediaProcessService.insert(task);
        return task;
    }



    /**
     * 更新关联信息
     */
    private void updateReferences(MediaProcess task, MediaFiles newFile) {
        // 1. 更新任务状态
        task.setStatus("2"); // 完成
        task.setFinishDate(LocalDateTime.now());
        mediaProcessService.updateById(task);

        // 2. 添加处理历史
        MediaProcessHistory history = new MediaProcessHistory();
        BeanUtils.copyProperties(task, history);
        mediaProcessHistoryService.insert(history);

        // 3. 更新教学计划中的引用
        Map<String, MediaFilesDto> updateMap = new HashMap<>();
        MediaFilesDto dto = new MediaFilesDto();
        BeanUtils.copyProperties(newFile, dto);
        updateMap.put(task.getFileId(), dto);
        teachplanMediaClient.updateFileInfo(updateMap);
    }

    /**
     * 删除临时文件
     */
    private void deleteTempFiles(File... files) {
        for (File file : files) {
            try {
                Files.deleteIfExists(file.toPath());
            } catch (IOException e) {
                log.warn("临时文件删除失败 | 文件名: {}", file.getName(), e);
            }
        }
    }

    /**
     * 处理异常情况
     */
    private void handleProcessingError(VideoProcessMessage message, Exception e) {
        VideoProcessMessageConver videoProcessMessageConver = new VideoProcessMessageConver();
        log.error("视频处理失败 | 文件ID: {}", message.getFileId(), e);
        // 更新任务状态为失败
        MediaProcessHistory selectMediaProcessHistory = mediaProcessHistoryService.selectByFileId(message.getFileId());
        MediaProcessHistory mediaProcessHistoryInit = videoProcessMessageConver.videoProcessMessageConverToMediaProcessHistory(message);//拷贝基本信息
        mediaProcessHistoryInit.setErrormsg(e.getMessage().length() > 500 ? e.getMessage().substring(0, 500) : e.getMessage());
        //非第一次失败
        if(selectMediaProcessHistory != null){
            mediaProcessHistoryInit.setId(selectMediaProcessHistory.getId());
            mediaProcessHistoryInit.setFailCount(selectMediaProcessHistory.getFailCount()+1);
            if(mediaProcessHistoryInit.getFailCount()>3){
                MediaFiles mediaFilesInsert = new MediaFiles();
                mediaFilesInsert.setAuditStatus("002002");      //跟该文件为待审核
                mediaFilesInsert.setFileId(mediaProcessHistoryInit.getFileId());
                mediaFilesService.updateNotEmptyByFileId(mediaFilesInsert);
            }
            mediaProcessHistoryService.updateById(mediaProcessHistoryInit);
            return;
        }
        mediaProcessHistoryInit.setFailCount(1);
        mediaProcessHistoryService.insert(mediaProcessHistoryInit);
    }


}
























